package org.funtester.performance.books.chapter03.section8;

import org.funtester.performance.books.chapter03.common.DataHandleTool;
import org.funtester.performance.books.chapter03.common.ThreadTool;

import java.util.Vector;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.LongAdder;

/**
 * TPS模型压测任务执行器
 */
public class TpsTaskExecutor {

    /**
     * 线程池
     */
    ThreadPoolExecutor poolExecutor;

    /**
     * 总执行次数
     */
    public static AtomicInteger executeNumStatistic;

    /**
     * 执行错误次数
     */
    public static AtomicInteger errorNumStatistic;

    /**
     * 用于记录所有请求时间
     */
    public static Vector<Integer> costTimeStatistic;

    /**
     * 任务描述
     */
    public String taskDesc;

    /**
     * tps模型压测任务
     */
    public TpsThreadTask task;

    /**
     * 开始时间,单位毫秒
     */
    public long startTimestamp;

    /**
     * 结束时间,单位毫秒
     */
    public long endTimestamp;

    /**
     * Rump-Up的总时间,单位秒
     */
    public int rumpUpTime;

    /**
     * 用于实时统计一段时间总耗时
     */
    public static LongAdder realTimeCostTime = new LongAdder();

    /**
     * 用于实时统计一段时间总次数
     */
    public static LongAdder realTimeCostTimes = new LongAdder();


    /**
     * 用于控制实时统计线程结束开关
     */
    public boolean realTimeKey = true;

    /**
     * TPS,即Rump-Up最大TPS
     */
    public int tps;

    /**
     * 测试总次数
     */
    public int totalTimes;

    /**
     * 实时信息输出线程,用于实时统计一段时间的TPS和平均耗时
     */
    public Thread realTimeThread;

    /**
     * 无参构造方法
     */
    public TpsTaskExecutor() {
    }

    /**
     * 构造方法
     * @param taskDesc 任务描述
     * @param rumpUpTime Rump-Up的总时间,单位秒
     */
    public TpsTaskExecutor(TpsThreadTask task, int tps, int rumpUpTime, int totalTimes, String taskDesc) {
        this.task = task;// 初始化任务
        this.taskDesc = taskDesc;// 初始化任务描述
        this.tps = tps;// 初始化TPS
        this.totalTimes = totalTimes;// 初始化测试总次数
        this.rumpUpTime = rumpUpTime;// 初始化Rump-Up时间
        this.executeNumStatistic = new AtomicInteger(0);// 初始化执行次数
        this.errorNumStatistic = new AtomicInteger(0);// 初始化执行错误次数
        this.costTimeStatistic = new Vector<>();// 初始化请求时间集合
        this.poolExecutor = ThreadTool.createPool(888, 1000); // 初始化线程池
    }

    /**
     * 开始执行任务
     */
    public void start() throws InterruptedException {
        if (realTimeThread == null) realTimeThread = new Thread() {// 创建线程,用于实时统计一段时间的TPS和平均耗时
            @Override
            public void run() {
                while (realTimeKey) {
                    ThreadTool.sleep(1000);// 休眠1秒
                    long sumCost = realTimeCostTime.sumThenReset();// 重置总耗时
                    long sumTimes = realTimeCostTimes.sumThenReset();// 重置总次数
                    System.out.println(String.format("实时统计TPS: %d, 平均耗时: %d", sumTimes, sumTimes == 0 ? 0 : sumCost / sumTimes));
                }
            }
        };
        realTimeThread.start();
        int step = tps / rumpUpTime;// 计算每秒提交的任务数量地增量
        while (true) {
            long runTime = (System.currentTimeMillis() - startTimestamp) / 1000;
            if (runTime >= rumpUpTime || TpsThreadTask.ABORT.get()) {// 判断是否结束Rump-Up
                System.out.println("Rump-Up结束,开始执行测试任务!");// 打印Rump-Up结束日志
                TpsThreadTask.countState = true;// 开始统计执行次数
                break;
            }
            long l = runTime * step;// 计算每秒提交的任务数量
            for (int i = 0; i < l; i++) {
                poolExecutor.execute(task);// 提交任务
            }
            ThreadTool.sleep(1000);// 休眠1秒
        }
        this.startTimestamp = System.currentTimeMillis();// 记录开始时间
        TEST:
        while (true) {
            for (int i = 0; i < tps; i++) {
                if (executeNumStatistic.get() >= totalTimes || TpsThreadTask.ABORT.get()) {// 判断是否结束测试任务
                    break TEST;
                }
                poolExecutor.execute(task);// 提交任务
            }
            ThreadTool.sleep(1000);// 休眠1秒
        }
        this.endTimestamp = System.currentTimeMillis();// 记录结束时间
        this.poolExecutor.shutdown();// 关闭线程池
        realTimeKey = false;// 关闭实时统计线程
        realTimeThread.join();// 等待实时统计线程结束
        handleData();// 处理数据
    }

    /**
     * 处理数据,在测试结束后运行
     */
    public void handleData() {
        long costTime = (endTimestamp - startTimestamp) / 1000;// 计算压测时长
        DataHandleTool.Quantile quantile = DataHandleTool.statisticData(costTimeStatistic);
        System.out.println(String.format("测试TPS: %d, 总执行次数: %d", executeNumStatistic.get() / costTime, executeNumStatistic.get()));// 打印统计的测试TPS
        quantile.print();// 打印请求时间统计结果
        System.out.println(String.format("任务执行完毕! 压测时长: %d 秒, 预期执行次数: %d, 实际执行次数 %d, 错误次数 %d, 耗时收集数量: %d", costTime, totalTimes, executeNumStatistic.get(), errorNumStatistic.get(), costTimeStatistic.size()));// 打印任务执行完毕
    }

    /**
     * 停止执行任务
     */
    public void stop() {
        TpsThreadTask.ABORT.set(true);// 设置终止标志
    }


}
